package com.intct.flink.study;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author gufg
 * @since 2025-10-13 14:10
 */
public class SQLTest03 {
    public static void main(String[] args) {
        // 配置属性
        Configuration conf = new Configuration();
        conf.set(RestOptions.BIND_PORT, "8081");

        // 1 获取取环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        // 并行度
        // env.setParallelism(1);
        // 开启检点
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        // 获取SQL执行环境
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        tenv.executeSql(
                "CREATE TABLE orders (\n" +
                        "     `user` BIGINT,\n" +
                        "     order_id BIGINT,\n" +
                        "     product STRING,\n" +
                        "     customer_id BIGINT,\n" +
                        "     order_time TIMESTAMP(3),\n" +
                        "     proc_time AS PROCTIME(), --处理时间\n" +
                        "     PRIMARY KEY(`user`) NOT ENFORCED\n" +
                        ") WITH (\n" +
                        "     'connector' = 'mysql-cdc',\n" +
                        "     'hostname' = 'cdh-node',\n" +
                        "     'port' = '13306',\n" +
                        "     'username' = 'root',\n" +
                        "     'password' = 'Test_090110',\n" +
                        "     'scan.startup.mode' = 'latest-offset',\n" +
                        "     'database-name' = 'm1',\n" +
                        "     'table-name' = 'orders'\n" +
                        ")"
        );

        String table = "CREATE TABLE customers_flink (\n" +
                " id INT,\n" +
                " name STRING,\n" +
                " country STRING,\n" +
                " zip STRING\n" +
                " ) WITH (\n" +
                " 'connector' = 'jdbc',\n" +
                " 'url' = 'jdbc:mysql://cdh-node:13306/m1?useSSL=false',\n" +
                " 'driver' = 'com.mysql.cj.jdbc.Driver',\n" +
                " 'username' = 'root',\n" +
                " 'password' = 'Test_090110',\n" +
                " 'lookup.cache.max-rows'='10000',  -- 缓存最大行数\n" +
                " 'lookup.cache.ttl'='1h',          -- 缓存 TTL（1 小时）\n" +
                " 'lookup.max-retries'='3',         -- 查询失败重试次数 \n" +
                " 'table-name' = 'customers'\n" +
                " )";

        tenv.executeSql(table);

        tenv.executeSql(
                "-- 结果表（Print 输出）\n" +
                        "CREATE TABLE order_product_result (\n" +
                        "    order_id STRING,\n" +
                        "    country STRING,\n" +
                        "    zip STRING \n" +
                        ") WITH (\n" +
                        "    'connector'='print'\n" +
                        ")");

//        tenv.executeSql("select * from customers_flink").print();

        tenv.executeSql("insert into order_product_result SELECT o.order_id, c.country, c.zip\n" +
                " FROM orders AS o\n" +
                " JOIN customers_flink FOR SYSTEM_TIME AS OF o.proc_time AS c\n" +
                " ON o.customer_id = c.id");

    }
}
